package org.tigr.htc.server.mw;

/*
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

import java.io.File;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.LineNumberReader;
import java.io.PrintWriter;
import java.util.Date;
import java.util.Hashtable;
import java.util.Iterator;
import java.util.Vector;

import org.apache.log4j.Logger;
import org.tigr.antware.shared.command.events.CommandRuntimeStatusLI;
import org.tigr.antware.shared.command.events.CommandStatusLI;
import org.tigr.antware.shared.command.events.FailureEvent;
import org.tigr.antware.shared.command.events.FinishEvent;
import org.tigr.antware.shared.command.events.InterruptEvent;
import org.tigr.antware.shared.command.events.ResumeEvent;
import org.tigr.antware.shared.command.events.StartEvent;
import org.tigr.antware.shared.command.events.SubmitEvent;
import org.tigr.antware.shared.command.events.SuspendEvent;
import org.tigr.htc.cmd.Command;
import org.tigr.htc.cmd.CommandConfig;
import org.tigr.htc.cmd.ICommand;
import org.tigr.htc.cmd.Param;
import org.tigr.htc.cmd.ParamType;
import org.tigr.htc.common.ConfigUpdateLI;
import org.tigr.htc.common.HTCConfig;
import org.tigr.htc.server.IRunner;
import org.tigr.htc.server.RunnerFactory;
import org.tigr.htc.server.condor.CondorUtils;

/**
 * The <b>Runner</b> class runs a command of type "mw". mw commands are array
 * jobs where instead of a single command running on the grid the command is a
 * template for an array of identical tasks where only the value of the
 * arguments change. The array of tasks can be based on a file containing the
 * arguments, a directory where each file in the directory is a argument, or a
 * simple argument that is a count.
 *
 */
public class Runner implements IRunner, CommandRuntimeStatusLI,
        CommandStatusLI, ConfigUpdateLI {
    static Logger log = Logger.getLogger(Runner.class);

    /**
     * <code> workerExe </code> the location of worker program.
     */
    static String workerExe = HTCConfig.getProperty("worker");

    /**
     * <code>cwd</code> system working directory for this runner. This is
     * where the runner stores logs, submit files, etc.
     */
    protected File cwd;

    /**
     * <code>cmd</code> the command to run.
     */
    protected Command cmd;

    /**
     * <code>workers</code> sub command created to run workers scripts.
     */
    protected Command workers;

    /**
     * <code>numWorkers</code> number of workers needed
     */
    protected int numWorkers;

    /**
     * <code> cmdHash </code> hash of the command object
     */
    protected Hashtable cmdHash = null;

    /**
     * <code>taskIndex</code> current index to hand out
     */
    protected int taskIndex = 0;

    /**
     * This constructor creates a new instance of MW runner to launch the
     * command.
     * 
     * @param p_cmd
     *           a <code>Command</code> object that will executed by this
     *           runner
     * @param p_cwd
     *           a <code>File</code> representing the parent directory
     */
    public Runner(Command p_cmd, File p_cwd) {
        log.debug("Runner constructor in dir " + p_cwd);

        cmd = p_cmd;

        log.debug("Create cwd directory");
        cwd = new File(p_cwd, Long.toString(cmd.getID()));

        if (!cwd.mkdirs()) {
            log.fatal("Could not make directory " + cwd);
        }

        // init mw information
        log.debug("Runner initialized with worker exec location "
                + workerExe);

        log.debug("Creating command hash and input/output");
        
        // create command hash
        cmdHash = new Hashtable();

        cmdHash.put("command", cmd.getCommand());
        cmdHash.put("initial_dir", cmd.getInitialDir());

        if (cmd.getOutput() != null) {
            cmdHash.put("output", cmd.getOutput());
        }
        if (cmd.getError() != null) {
            cmdHash.put("error", cmd.getError());
        }
        if (cmd.getInput() != null) {
            cmdHash.put("input", cmd.getInput());
        }
        // Vector to hold parameters
        Vector p = new Vector();

        log.debug("Create parameters");
        if (cmd.getParamsCount() > 0) {
            Iterator params = cmd.getParams().iterator();

            for (; params.hasNext();) {
                Param param = (Param) params.next();
                String paramStr = param.getType().toString() + "=";

                if (param.getKey() != null) {
                    paramStr += param.getKey();
                }

                paramStr += "=";

                paramStr += param.getValue();
                log.debug("Add param string " + paramStr);
                p.add(paramStr);
            }
        }

        cmdHash.put("params", p);
    }

    /*
     * Interface to Worker
     */

    /**
     * The method <code>getCommandHash</code>
     * This is used to initialize the worker
     * 
     * @return cmdHash, a Hashtable
     */
    public Hashtable getCommandHash() {
        log.debug("In getCommandHash " + cmd.getID());
        return cmdHash;
    }

    // Send the worker the next index to use
    public int getNextIndex(int worker, String machine) {
        int returnTask = -1;

        synchronized (this) {
            if (taskIndex < cmd.getNumTasks()) {
                log.debug("There are more tasks for cmd " + cmd.getID());
                log.debug("current task " + taskIndex);
                returnTask = taskIndex;
                log.debug("Cmd Task start time :" + new Date());
                log.debug("Cmd Task machine    :" + machine);
                log.debug("Cmd Task index      :" + taskIndex);
                log.debug("Cmd Task worker     :" + worker);
                cmd.taskStarted("Task started ", new Date(), machine,
                        taskIndex, worker);
                taskIndex++;
                log.debug("Up Task index for cmd " + cmd.getID() + " to "
                        + taskIndex);
            } else {
                log.debug("Cmd " + cmd.getID() + " has no more tasks.");
            }
        }

        log.debug("Returning Index for cmd " + cmd.getID() + " is " + returnTask);

        return returnTask;
    }

    //
    // task is finished
    //
    public void setTaskFinished(int index, int value, String message) {
        try {
            log.debug("Return value for task " + index + " was " + value);
            cmd.taskFinished(message, new Date(), value, index);
        } catch (Exception e) {
            log.debug("Failure in setTaskFinished: " + e.getMessage(), e);
        }
    }

    //
    // task failed
    //
    public int setTaskFailure(int index, String message) {
        try {
            log.debug("Failure for task " + index + " was " + message);
            cmd.taskFailure(new Exception(message), index);
            return 0;
        } catch (Exception e) {
            log.debug("Failure in setTaskFailure: " + e.getMessage(), e);
            return -1;
        }
    }

    /**
     * run starting point take a command object and create a worker command and launch the worker
     * command on the grid
     * 
     * throws invalid command object
     *  
     */
    public void run() {
        try {
            log.debug("In run.");
            cmd.getConfig().setLogLocation(cwd.getAbsolutePath());
            log.debug("Submit URL is :" + cmd.getSubmitURL());
            // calculate tasks and number of tasks per worker
            log.info("Init tasks");

            initTasks();
            
            // Fire task submitted event
            cmd.taskSubmitted(cmd.getSubmitURL(), cwd.getAbsolutePath()+ "/worker.log");
            
            // create worker cmd
            log.info("Create worker command.");
            workers = createWorkerCmd();
        } catch (Exception e) {
            log.error("Error with Runner " + e.getMessage());
            cmd.fireFailureEvent(e);
            log.error("Update failure command.");
            cmd.updateCommand();

            // This catch block could be caused by the inability to create
            // the worker so firing a failure event causes a null pointer exception
            // also in this situation what is the point in updating the command status
            // as the worker status is not event set till the createCommand is called
            // later. Hence commenting out, but make sure. AAM
            // workers.fireFailureEvent(e);
            // workers.updateCommand();
            return;
        }

        try {
            // add listener
            workers.addRuntimeStatusListener(this);
            workers.addStatusListener(this);

            log.info("Persist sub-grid cmd for MW cmd");
            workers.createCommand();

            // submit request for workers
            startWorkerCmd();
            // bye - wait for workers to contact us
        } catch (Exception e) {
            log.debug("Exception in", e);
        }
    }

    /**
     * The method <code>startWorkerCmd</code> launches the worker command on the grid
     * 
     */
    public void startWorkerCmd() {
        IRunner runner = null;
        log.debug("In startWorkerCmd");

        try {
            log.debug("Creating the runner");
            runner = RunnerFactory.newRunner(workers, cwd);

            log.debug("Starting the runner");
            runner.run();
            cmd.setGridJobID(((Command) runner.getICommand()).getGridJobID());

        } catch (Exception e) {
            log.error("Exception in startWorkerCmd " + e.getMessage());
            // fire failed event
            cmd.fireFailureEvent(e);
            workers.fireFailureEvent(e);
            cmd.updateCommand();
            workers.updateCommand();
        }
    }

    // Calculate the number of jobs and workers needed
    // take the MIN(cmd.times, smallest iterator parameter)
    // need to loop over each parameter and count
    // have to use a external Perl script to count because of permissions(use // sudo)
    public void initTasks() throws Exception {
        log.debug("Creating the param file: " + cwd + "/params");
        File paramFile = new File(cwd, "params");
        PrintWriter pf = new PrintWriter(new FileWriter(paramFile), true);

        File workerLog = new File(cwd, "worker.log");
        if (!workerLog.createNewFile()) {
            log.error("Could not create a new file on: "
                    + workerLog.getAbsolutePath());
        }

        int numIterParams = 0;

        // Go over the param vector and print out any iter Params
        if (cmd.getParamsCount() > 0) {
            Iterator params = cmd.getParams().iterator();

            for (; params.hasNext();) {
                Param param = (Param) params.next();
                if ((param.getType() != ParamType.PARAM)
                        && (param.getType() != ParamType.UNKNOWN)) {
                    String paramStr = param.getType().toString() + ":";

                    paramStr += param.getValue();

                    numIterParams++;
                    log.debug("Add param str " + paramStr
                            + "\n number of params " + numIterParams);
                    pf.println(paramStr);
                }
            }
        }

        pf.close();

        int numTasks = 1;
        int ltimes = cmd.getTimes();

        if (numIterParams > 0) {
            log.debug("iterParams count " + numIterParams);
            numTasks = CondorUtils.initTasks(paramFile);
            if (numTasks == 0) {
                throw new Exception("Command Iterative Parameter contained "
                        + "zero values");
            }
        }

        if ((ltimes > 0) && (numTasks > ltimes)) {
            numTasks = ltimes;
        }

        // Set numtasks to the Command object
        cmd.setNumTasks(numTasks);
        cmd.getStatus().setWaiting(numTasks);

        int maxWorkers = cmd.getMaxWorkers();

        if ((maxWorkers > 0) && (numTasks > maxWorkers)) {
            numTasks = maxWorkers;
        }

        // figure out workers
        if (numTasks < 70) {
            numWorkers = numTasks;
        } else {
            numWorkers = numTasks / 10;
        }

        // Limit the numWorkers to approximately half of the cluster
        // capacity. This value should come from the server configuration file.
        int maxNumWorkers = Integer.parseInt(HTCConfig.getProperty(
                "max_num_array_job_workers").toString());

        log.debug("Maximum number of workers should be " + maxNumWorkers);

        if (numWorkers > maxNumWorkers) {
            numWorkers = maxNumWorkers;
        }

        log.info("Number of tasks initialized with: " + numTasks);
    }

    /**
     * The method <code>createWorkerCmd</code> creates the worker command that
     * will launched on the grid to process the master worker command request
     * submitted by the user.
     * 
     * @return a <code>Command</code> object representing the child command.
     */
    public Command createWorkerCmd() {
        log.debug("In createWorkerCmd.");

        try {
            workers = new Command();
            workers.setID(Command.createCmdID());
            // set command/output/group/user specific to worker command
            workers.setCommand(workerExe);
            workers.setType("internal");
            // set initial dir set to /
            workers.setInitialDir("/");

            //workers.sets
            File workerOut = new File(cwd, "worker.out");
            if (!workerOut.createNewFile()) {
                log.warn("Could not create a new file at: "
                        + workerOut.getAbsolutePath());
            }

            // Set error path
            workers.setError(workerOut.getAbsolutePath());

            // Set output path
            workers.setOutput(workerOut.getAbsolutePath());

            // Set setTimes to number workers to be spawned.
            workers.setTimes(numWorkers);

            // Set command group
            workers.setGroup(cmd.getGroup());

            // Set the user name
            workers.setUserName(cmd.getUserName());

            //  1 - master address
            workers.addParam("-m " + Server.MWURI);

            // Add the index number, to be replaced by the SGERunner
            workers.addParam("-s $(Index)");

            // Add command id
            workers.addParam("-c " + cmd.getID());

            // Add the worker log location
            workers.addParam("-l " + cwd + "/worker.log.$(Index)");
            
            // Set monitor host for this command 
            workers.setMonitorHost(cmd.getMonitorHost());

            // copy over requirements to worker
            CommandConfig wconfig = workers.getConfig();
            CommandConfig cconfig = cmd.getConfig();
            
            if (cconfig != null) {
                wconfig.setOpSys(cconfig.getOpSys());
                wconfig.setLength(cconfig.getLength());
                wconfig.setCmdClass(cconfig.getCmdClass());
                wconfig.setMemory(cconfig.getMemory());
                wconfig.setPassThrough(cconfig.getPassThrough());
                // We make all the MW jobs as non-evictable until 
                // the issue with resuming MW jobs is fixed.
                wconfig.setEvictable(false);
                
                // Set running time if specified, any
                if (cconfig.getRunningTime() > 0) {
                    wconfig.setRunningTime(cconfig.getRunningTime());
                }
                log.debug("Check if we need to set the environment "
                        + cconfig.isGetenv());
                if (cconfig.isGetenv()) {
                    log.debug("Setting getenv to true and transferring vector");
                    workers.setEnv(cmd.getEnv());
                    wconfig.setGetenv(true);
                }
            }

        } catch (Exception e) {
            log.debug("Exception creating worker for command with id: "
                    + cmd.getID(), e);
            return null;
        }

        return workers;
    }

    public ICommand getICommand() {
        return cmd;
    }

    // CommandStatusLI methods
    public void commandStarted(StartEvent event) {
        log.debug("Command " + event.getCommandID() + " started");
        String gridId = ((Command) event.getSource()).getGridJobID();

        cmd.setGridJobID(gridId);

        cmd.updateCommand();
    }

    public void commandRestarted(StartEvent event) {
        log.debug("Command #" + event.getCommandID() + " restarted");
    }

    public void commandInterrupted(InterruptEvent event) {
    	log.debug("Command #" + event.getCommandID() + " interrupted");
        
        // Merge individual worker logs
    	mergeWorkerLogs();
    }

    public void commandFailed(FailureEvent event) {
        log.debug("Got a failure event for Command " + event.getCommandID());
        
        // Merge individual worker logs
        mergeWorkerLogs();
        
        cmd.taskFailure(event.getFailure(), taskIndex);
    }

    // Need to check that we are done also
    public void commandFinished(FinishEvent event) {
        log.debug("Got a finish event for " + event.getCommandID());

        // Fire finish event
        log.debug("Firing taskFinished for Command with id " + cmd.getID());  
        
        // Merge individual worker logs
        mergeWorkerLogs();
        
        cmd.taskFinished(event.getMessage(), event.getDate(), event.getReturnValue(), taskIndex);
    }

    // CommandRuntimeStatusLI methods
    public void commandSubmitted(SubmitEvent event) {
        log.debug("Got a submitted event for " + event.getMessage());
    }

    public void commandResumed(ResumeEvent event) {
        log.debug("Got a resume event for " + event.getMessage());
    }

    public void commandSuspended(SuspendEvent event) {
        log.debug("Got a suspend event for " + event.getMessage());
    }

    public boolean stop() {
        return false;
    }

    public boolean hold() {
        return false;
    }

    public boolean release() {
        return false;
    }

    public boolean isRunnerComplete() {
        return true;
    }

    public void configUpdated() {
        workerExe = HTCConfig.getProperty("worker");
    }
    
    public void mergeWorkerLogs() {
    	String line = null;
		FileOutputStream fos = null;
    	try {
			fos = new FileOutputStream(cwd + "/worker.log", true);
			log.debug("Merging the individual Worker log files for "
					+ cmd.getID() + " to " + cwd + "/worker.log");
			for (int i = 1; i <= cmd.getNumTasks(); i++) {
				LineNumberReader fileReader = new LineNumberReader(
						new FileReader(new File(cwd + "/worker.log." + i)));
				while ((line = fileReader.readLine()) != null) {
					fos.write(line.getBytes());
					fos.write("\n".getBytes());
				}
				fileReader.close();
			} // end for loop
			
			// Close FileoutputStream
			fos.flush();
			fos.close();
		} catch (Exception e) {
			// TODO: handle exception
			log.warn("Could not merge the individual worker log files " + e.getMessage());
		}
    	log.debug("Finished merging the individual worker log files for " + cmd.getID());
    }
}
